package com.example.kafka.fast;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.example.kafka.constant.KafkaConstant;
import com.example.kafka.constant.StreamConstant;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KeyValueMapper;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Reducer;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.kstream.internals.WindowedSerializer;

import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.TimeUnit;


/**
 * @author Liyuhang
 */
public class KafkaStreamDemo1 {


    protected static Properties  getDefaultProperties(){
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-stream-demo_stream_aid");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "47.100.82.240:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
        return props;
    }

    protected static void streamProcess( StreamsBuilder builder){
        //接收消息
        KStream<String, String> source  = builder.stream(StreamConstant.IN_PUT_TOPIC);
        source.selectKey(new KeyValueMapper<String, String, String>() {
                    @Override
                    public String apply(String key, String value) {
                        return "timeStamp";
                    }
                }).map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
                    /**
                     * 把消息的key,重新赋值，
                     * @param key
                     * @param value
                     * @return
                     */
                    @Override
                    public KeyValue<String, String> apply(String key,String value) {
                        String[] strings = value.split("-");
                        return new KeyValue<>(value,strings[1]);
                    }
                })
                .groupByKey()
                .windowedBy(TimeWindows.of(TimeUnit.SECONDS.toMillis(2000)))
                .reduce(new Reducer<String>() {
                    @Override
                    public String apply(String value1, String value2) {
                        System.out.println("value1=" + value1+ ", value2=" + value2);
                        JSONObject json = JSON.parseObject(value1);
                        Integer temperature = json.getInteger("temp");
                        if (temperature > Integer.parseInt(value2)) {
                            return temperature.toString();
                        }
                        else {
                            return value2;
                        }
                    }
                })
                .toStream()
                //把处理后的key和value转换String
                .map((key, value) -> {
                    return new KeyValue<>(key.key().toString(), value.toString());
                })
                //发送消息
                .to("out_topic");

    }


}
